package cn.jly.flink.quickstart;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;

/**
 * flink quick start
 *
 * @author lanyangji
 * @date 2020/8/14
 */
public class StreamWordCount {
    private static final Logger LOGGER = LoggerFactory.getLogger(StreamWordCount.class);

    public static void main(String[] args) throws Exception {
        LOGGER.info("hello");
        LOGGER.warn("world");
        if (args.length < 2) {
            LOGGER.error("args length < 2 : {}", Arrays.toString(args));
            return;
        }

        String hostname = args[0].trim();
        int port = Integer.parseInt(args[1]);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.socketTextStream(hostname, port)
                .flatMap(new LineSplitMapFunction("pre"))
                .keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                        return stringIntegerTuple2.f0;
                    }
                })
                .sum(1)
                .printToErr();

        env.execute("StreamWordCount");
    }
}
